package com.margo.project.consumer.kafka.consumer.configuration;

import org.springframework.beans.factory.ObjectProvider;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.boot.autoconfigure.kafka.ConcurrentKafkaListenerContainerFactoryConfigurer;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.support.converter.StringJsonMessageConverter;


/**
 * 
 * 接受消息报错
 * 1)Caused by: java.lang.IllegalStateException: 
 * No group.id found in consumer config, container properties, or @KafkaListener annotation; 
 * a group.id is required when group management is used.
 * 
 * 2)org.springframework.messaging.converter.MessageConversionException: 
 * Cannot handle message; nested exception is org.springframework.messaging.converter.MessageConversionException: 
 * Cannot convert from [java.lang.String] to 
 * [com.margo.project.consumer.common.model.EasydlDiscernMessageVO] for GenericMessage
 * @author fxb208
 *
 */
@Configuration
public class KafkaConsumerConfiguration {
	
	@Autowired
	private KafkaProperties properties;
	
	@Bean
	public StringJsonMessageConverter jsonConverter() {
	        return new StringJsonMessageConverter();
	}
	
	@Bean
	public ConcurrentKafkaListenerContainerFactory<?, ?> kafkaListenerContainerFactory(
			ConcurrentKafkaListenerContainerFactoryConfigurer configurer,
			ObjectProvider<ConsumerFactory<Object, Object>> kafkaConsumerFactory) {
		ConcurrentKafkaListenerContainerFactory<Object, Object> factory = new ConcurrentKafkaListenerContainerFactory<>();
		configurer.configure(factory, kafkaConsumerFactory
				.getIfAvailable(() -> new DefaultKafkaConsumerFactory<>(this.properties.buildConsumerProperties())));
		configureContainer(factory.getContainerProperties());
		return factory;
	}

	//
	private void configureContainer(ContainerProperties containerProperties) {
		containerProperties.setSyncCommits(false);	
	}
}
